package org.platform.processor.monitor.job;

import com.mc.mic.core.tool.helper.DiscoverHelper;
import lombok.extern.slf4j.Slf4j;
import org.platform.mo.bussmo.diagnose.event.DeviceBO;
import org.platform.processor.monitor.data.CacheLibraryPO;
import org.platform.tool.util.JsonUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.zeromq.ZMQ;

import javax.annotation.PostConstruct;
import java.util.HashSet;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * 第一步: 该类主要用于接收zeroMQ发送过来的设备属性
 * 1.通过线程动态维护需要接收服务器的IP
 *
 * @author FengJie
 * @date 2018/5/15
 */
@Service
@Slf4j
public class ReceiveDeviceForMQJob {
    private static final String CONNECT_TYPE = "tcp://";
    private static final String CONNECT_PORT = ":28887";
    private static final String TOPIC = "DeviceStatus";
    private static final int CYCLE = 5;

    @Autowired
    private ScheduledExecutorService scheduler;
    @Autowired
    private DiscoverHelper discoverHelper;

    @PostConstruct
    private void init() {
        ZMQ.Context context = ZMQ.context(1);
        ZMQ.Socket socket = context.socket(ZMQ.SUB);
        socket.subscribe(TOPIC);

        connect(socket);
        new Receive(socket).start();
    }

    /**
     * 从eureka服务中心上获取所有实例IP,然后动态绑定
     * 解决了服务掉线然后又上线的问题
     */
    private void connect(ZMQ.Socket socket) {
        //存放已经连接的ip
        Set<String> ipList = new HashSet<>();

        scheduler.scheduleAtFixedRate(() -> {
            Set<String> instanceIps = discoverHelper.getAllInstanceIps();
            instanceIps.forEach(instanceIp -> {
                if (!ipList.contains(instanceIp)) {
                    socket.connect(CONNECT_TYPE + instanceIp + CONNECT_PORT);
                    ipList.add(instanceIp);
                }
            });
        }, 1, 10, TimeUnit.SECONDS);

    }

    class Receive extends Thread {
        Queue<DeviceBO> deviceQueue;
        ZMQ.Socket socket;

        Receive(ZMQ.Socket socket) {
            this.socket = socket;
            this.deviceQueue = CacheLibraryPO.getCacheLibraryPO().getDeviceQueue();
        }

        @Override
        @SuppressWarnings("InfiniteLoopStatement")
        public void run() {
            while (true) {
                try {
                    //第一次接收到的是主题  第二次接收到的是内容
                    String anObject = socket.recvStr(0);
                    if (TOPIC.equals(anObject)) {
                        String device = socket.recvStr(0);
                        DeviceBO deviceBO = JsonUtils.jsonToBean(device, DeviceBO.class, true);
                        deviceQueue.offer(deviceBO);
                    }
                } catch (Exception e) {
                    log.error("接收设备参数异常,异常信息:", e);
                }
            }
        }
    }
}
